package com.it.state;

import com.it.operator.utils.SourceUtils;
import com.it.pojo.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 每10s统计一次每个用户的点击次数，使用事件时间进行驱动.
 *
 * @author code1997
 */
public class PeriodicPvDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        executionEnvironment.setParallelism(1);
        SingleOutputStreamOperator<Event> eventStream = SourceUtils.getEventSource(executionEnvironment).assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO).withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
            @Override
            public long extractTimestamp(Event element, long recordTimestamp) {
                return element.timestamp;
            }
        }));
        eventStream.print("input");
        //每个用户的访问量
        eventStream.keyBy(data -> data.user).process(new PeriodicPvCompute()).print();
        executionEnvironment.execute();
    }

    public static class PeriodicPvCompute extends KeyedProcessFunction<String, Event, String> {

        ValueState<Long> countState;
        //是否存在定时器
        ValueState<Long> haveTimer;

        @Override
        public void open(Configuration parameters) throws Exception {
            countState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("countState", Types.LONG));
            haveTimer = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timerState", Types.LONG));
        }

        @Override
        public void processElement(Event value, KeyedProcessFunction<String, Event, String>.Context ctx, Collector<String> out) throws Exception {
            Long oldCount = countState.value();
            countState.update(oldCount == null ? 1 : oldCount + 1);
            if (haveTimer.value() == null) {
                //需要定义一个10s的定时器
                ctx.timerService().registerEventTimeTimer(value.timestamp + 5 * 1000L);
                haveTimer.update(value.timestamp + 5 * 1000L);
            }
        }

        @Override
        public void onTimer(long timestamp, KeyedProcessFunction<String, Event, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
            //定时器触发，输出一次统计结果
            out.collect("时间：" + ctx.getCurrentKey() + "的记录数为：" + countState.value());
            //清空状态
            haveTimer.clear();
            ctx.timerService().registerEventTimeTimer(timestamp + 5 * 1000L);
            haveTimer.update(timestamp + 5 * 1000L);
        }
    }
}
